package com.study.flink

import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.api.windowing.time.Time

/**
  *
  * @author: stephen.shen
  * @create: 2019-05-31 10:04
  */
object Test {

  def main(args: Array[String]): Unit = {
    // 1 获取执行环境
    val env = StreamExecutionEnvironment.getExecutionEnvironment

    // 2 获取输入数据
    import org.apache.flink.api.scala._
    val dataStream = env.socketTextStream("localhost",9999)

    // 3 Transformation
    val resultStream = dataStream.flatMap(_.split(" "))
      .map((_, 1))
      .keyBy(_._1)
      .timeWindow(Time.seconds(10))
      .reduce((x, y) => (x._1, x._2 + y._2))

    // 4 输出
    resultStream.print()

    // 5 启动任务
    env.execute("Demo")
  }
}
